package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/louloulin/dataflare/pkg/client/go"
)

func main() {
	// Get API URL and key from environment variables
	apiURL := getEnv("DATAFLARE_API_URL", "http://localhost:8080")
	apiKey := getEnv("DATAFLARE_API_KEY", "")

	// Create client
	c := client.NewClient(
		apiURL,
		client.WithAPIKey(apiKey),
		client.WithTimeout(30*time.Second),
	)

	// Create context
	ctx := context.Background()

	// Create workflow
	workflow, err := c.CreateWorkflow(ctx, &client.CreateWorkflowRequest{
		Name:        "example-workflow",
		Description: "Example workflow",
		Definition: `
version: "1.0"
name: "example-workflow"
description: "Example workflow"
config:
  mode: "stream"
  execution:
    engine: "actor"
    parallelism: 1
  state:
    backend: "memory"
schemas:
  record:
    fields:
      id:
        type: "string"
        required: true
      value:
        type: "number"
        description: "Value"
sources:
  memory_source:
    type: "memory"
    schema: "record"
    config:
      data: [
        {"id": "1", "value": 10},
        {"id": "2", "value": 20},
        {"id": "3", "value": 30}
      ]
    description: "In-memory data source"
sinks:
  console_sink:
    type: "console"
    schema: "record"
    config:
      format: "json"
    description: "Console output sink"
nodes:
  process:
    type: "transform"
    description: "Process input records"
    inputs:
      default:
        schema: "record"
    outputs:
      default:
        schema: "record"
    transform:
      language: "javascript"
      code: |
        function process(record) {
          return {
            id: record.id,
            value: record.value * 2
          };
        }
edges:
  - from:
      source: "memory_source"
    to:
      node: "process"
      input: "default"
  - from:
      node: "process"
      output: "default"
    to:
      sink: "console_sink"
`,
	})
	if err != nil {
		fmt.Printf("Failed to create workflow: %v\n", err)
		os.Exit(1)
	}

	fmt.Printf("Created workflow: %s\n", workflow.ID)

	// Run workflow
	run, err := c.RunWorkflow(ctx, workflow.ID, &client.RunWorkflowRequest{
		Parameters: map[string]interface{}{
			"param1": "value1",
		},
	})
	if err != nil {
		fmt.Printf("Failed to run workflow: %v\n", err)
		os.Exit(1)
	}

	fmt.Printf("Started workflow run: %s\n", run.ID)

	// Wait for workflow to complete
	for {
		// Get workflow run
		run, err := c.GetWorkflowRun(ctx, workflow.ID, run.ID)
		if err != nil {
			fmt.Printf("Failed to get workflow run: %v\n", err)
			os.Exit(1)
		}

		// Check if workflow is completed
		if run.Status == "completed" || run.Status == "failed" || run.Status == "cancelled" {
			fmt.Printf("Workflow run completed with status: %s\n", run.Status)
			if run.Error != "" {
				fmt.Printf("Error: %s\n", run.Error)
			}
			break
		}

		// Wait for a while
		fmt.Printf("Workflow run status: %s\n", run.Status)
		time.Sleep(1 * time.Second)
	}

	// List workflow runs
	runs, err := c.ListWorkflowRuns(ctx, workflow.ID, "", 10)
	if err != nil {
		fmt.Printf("Failed to list workflow runs: %v\n", err)
		os.Exit(1)
	}

	fmt.Printf("Workflow runs:\n")
	for _, run := range runs.Runs {
		fmt.Printf("- %s: %s\n", run.ID, run.Status)
	}

	// Delete workflow
	err = c.DeleteWorkflow(ctx, workflow.ID)
	if err != nil {
		fmt.Printf("Failed to delete workflow: %v\n", err)
		os.Exit(1)
	}

	fmt.Printf("Deleted workflow: %s\n", workflow.ID)
}

// getEnv gets an environment variable or returns a default value
func getEnv(key, defaultValue string) string {
	value := os.Getenv(key)
	if value == "" {
		return defaultValue
	}
	return value
}
